package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.bean.UserRegisterBean;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.KafkaUtil;
import org.apache.commons.collections.MapUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.atomic.AtomicInteger;

public class DwsUserUserRegisterWindow {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        String topic_name = "topic_db";
        String groupId = "dws_user_user_register_window";
        /**
         *{
         *     "database":"edu",
         *     "table":"vip_change_detail",
         *     "type":"insert",
         *     "ts":1592100616,
         *     "xid":73516,
         *     "commit":true,
         *     "data":{
         *         "id":10304,
         *         "user_id":34,
         *         "from_vip":3,
         *         "to_vip":4,
         *         "create_time":"2022-09-03 10:10:17"
         *     }
         * }
         */
        SingleOutputStreamOperator<JSONObject> map = env
                //读取kafka
                .addSource(KafkaUtil.getKafkaConsumer(topic_name, groupId), TypeInformation.of(String.class))
                // 转化
                .map(new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String value) throws Exception {
                        return JSON.parseObject(value);
                    }
                });
        SingleOutputStreamOperator<JSONObject> filter = map
                //过滤
                .filter(new FilterFunction<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        String database = MapUtils.getString(value, "database");
                        String table = MapUtils.getString(value, "table");
                        String type = MapUtils.getString(value, "type");
                        return "edu".equals(database) && "user_info".equals(table) && "insert".equals(type);
                    }
                });

        SingleOutputStreamOperator<UserRegisterBean> streamOperator =
                filter
                        //设置watermark
                        .assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner(
                                new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                        String createTime = element.getJSONObject("data").getString("create_time");
                                        return DateFormatUtil.toTs(createTime, true);
                                    }
                                }
                        ))
                        //滚动窗口计算
                        .windowAll(TumblingEventTimeWindows.of(Time.seconds(10)))
                        .process(new ProcessAllWindowFunction<JSONObject, UserRegisterBean, TimeWindow>() {
                            @Override
                            public void process(ProcessAllWindowFunction<JSONObject, UserRegisterBean, TimeWindow>.Context context,
                                                Iterable<JSONObject> elements, Collector<UserRegisterBean> out) throws Exception {

                                long start = context.window().getStart();
                                long end = context.window().getEnd();
                                String stt = DateFormatUtil.toYmdHms(start);
                                String edt = DateFormatUtil.toYmdHms(end);

                                AtomicInteger count = new AtomicInteger();

                                elements.forEach(item -> count.getAndIncrement());


                                UserRegisterBean userRegisterBean = new UserRegisterBean(stt, edt, count.longValue(), System.currentTimeMillis());
                                out.collect(userRegisterBean);
                            }
                        });

        streamOperator.print("reduce");

        env.execute();
    }
}
